//package com.xiguaji.tiktok.task.awemecomment.event;
//
//import cn.hutool.core.date.DateUtil;
//import com.fasterxml.jackson.databind.JavaType;
//import com.fasterxml.jackson.databind.ObjectMapper;
//import com.google.common.eventbus.AllowConcurrentEvents;
//import com.google.common.eventbus.Subscribe;
//import com.gugee.tiktok.common.model.dal.blogger.AwemeCommentRepository;
//import com.gugee.tiktok.common.model.es.Keyword;
//import com.gugee.tiktok.common.model.mysql.Aweme;
//import com.gugee.tiktok.common.model.mysql.AwemeComment;
//import com.gugee.tiktok.common.model.spider.comment.AwemeCommentResponse;
//import com.xiguaji.boot.eventbus.Event;
//import com.xiguaji.boot.eventbus.EventBusTemplate;
//import com.xiguaji.boot.eventbus.IEventListener;
//import com.xiguaji.boot.filecenter.s3.S3Template;
//import com.xiguaji.boot.toolkit.JsonUtil;
//import com.xiguaji.boot.toolkit.ThreadPoolUtils;
//import com.xiguaji.tiktok.task.aweme.client.BloggerClient;
//import com.xiguaji.tiktok.task.aweme.event.ESRefreshAwemeEventListener;
//import com.xiguaji.tiktok.task.aweme.event.SaveAwemeKeywordEventListener;
//import com.xiguaji.tiktok.task.awemecomment.spider.AwemeCommentSpiderClient;
//import com.xiguaji.tiktok.task.core.ScheduleTemplate;
//import com.xiguaji.tiktok.task.monitor.AwemesComentStatics;
//import com.xxl.job.core.log.XxlJobLogger;
//import lombok.Getter;
//import lombok.Setter;
//import lombok.SneakyThrows;
//import lombok.extern.slf4j.Slf4j;
//import lombok.val;
//import org.elasticsearch.action.get.GetRequest;
//import org.elasticsearch.action.get.GetResponse;
//import org.elasticsearch.client.RequestOptions;
//import org.elasticsearch.client.RestHighLevelClient;
//import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.context.annotation.Lazy;
//import org.springframework.stereotype.Component;
//import org.springframework.util.CollectionUtils;
//
//import javax.annotation.PostConstruct;
//import java.io.IOException;
//import java.util.*;
//import java.util.concurrent.ConcurrentHashMap;
//import java.util.concurrent.ForkJoinPool;
//import java.util.concurrent.atomic.AtomicInteger;
//
//@Component
//@Slf4j
//public class SaveAwemeCommentEventListener implements IEventListener<SaveAwemeCommentEventListener.SaveAwemeCommentEvent> {
//    private final String AWEME_INDEX = "aweme_index_";
//
//
//    @Autowired
//    AwemeCommentSpiderClient awemeCommentSpiderClient;
//
//    @Autowired
//    AwemeCommentRepository awemeCommentRepository;
//
//    @Autowired
//    AwemeComment.AwemeCommentMapper awemeCommentMapper;
//
//    @Autowired
//    ScheduleTemplate scheduleTemplate;
//
//    @Autowired
//    RestHighLevelClient highLevelClient;
//    private String s3workName = "AwemeCommentStaticS3-worker";
//    private String workName = "SaveAwemeComment-worker";
//    private final int collectMaxComments = 200;
//    @Autowired
//    @Lazy
//    EventBusTemplate eventBusTemplate;
//
//    ForkJoinPool forkJoinPool = ThreadPoolUtils.forkJoinPool(60, workName);
//    ForkJoinPool s3ForkJoinPool = ThreadPoolUtils.forkJoinPool(200, workName);
//    private static final ObjectMapper MAPPER = new ObjectMapper();
//    @PostConstruct
//    public void init() {
//        scheduleTemplate.attach(workName, forkJoinPool);
//        scheduleTemplate.attach(s3workName, s3ForkJoinPool);
//    }
//
//    @SneakyThrows
//    @Override
//    @Subscribe
//    @AllowConcurrentEvents
//    public void handle(SaveAwemeCommentEvent event) {
//        long start = System.currentTimeMillis();
//        ConcurrentHashMap<Long, Byte> removeAwemes = new ConcurrentHashMap<>();
//        ConcurrentHashMap<Aweme, List<AwemeComment>> gt200commentsAwemes = new ConcurrentHashMap<>();
//        Byte zero = (byte) 0;
//        s3ForkJoinPool.submit(() -> event.awemes.parallelStream().forEach(a -> {
//            try {
//                //视频已有评论数小于50则先不爬取
//                if (a.getOCommentCount() < 50) {
//                    removeAwemes.put(a.getOAwemeId(), zero);
//                    return;
//                }
//                //s3中存的评论数
//                List<AwemeComment> comments = getCommentsFromS3(a.getOAwemeId());
//                int s3Comments = comments.size();
//                //视频已收录评论数大于1000则不再爬取
//                if (s3Comments >= collectMaxComments) {
//                    removeAwemes.put(a.getOAwemeId(), zero);
//                    //收集评论大于指定数的视频
//                    gt200commentsAwemes.put(a, comments);
//                    //todo 未能考虑关键词数量》0且需要更新关键词的情形
////                   if(findKeywordByES(a).size() <= 0) {
////                       gt200commentsAwemes.put(a.getOAwemeId(),comments);
////                   }
//                    return;
//                }
//                //评论数增量少于100则先不爬取
//                if (a.getOCommentCount() - s3Comments < 100) {
//                    removeAwemes.put(a.getOAwemeId(), zero);
//                }
//            } catch (Exception ex) {
//                log.error("SaveAwemeComment error.过滤掉不满足评论采集条件的视频 awemeId:" + a.getOAwemeId(), ex);
//
//            }
//        })).get();
//
//        event.awemes.removeIf(a -> removeAwemes.containsKey(a.getOAwemeId()));
//        event.awemes.removeIf(gt200commentsAwemes::containsKey);
//
//        log.info("被过滤视频数={}，最后需处理视频数={}", removeAwemes.size(), event.awemes.size());
//
//
//        event.statics.getTotalAwemes().addAndGet(event.awemes.size());
//        XxlJobLogger.log("SaveAwemeCommentEvent real start awemeSize :{}", event.awemes.size());
//        log.info("SaveAwemeCommentEvent real start awemeSize :{}", event.awemes.size());
//
//        if (event.awemes.size() == 1) {
//            run(event.awemes.get(0), event);
//        } else {
//            forkJoinPool.submit(() -> event.awemes.parallelStream().forEach(aweme -> {
//                try {
//                    run(aweme, event);
//                } catch (Exception ex) {
//                    log.error("forkJoinPool SaveAwemeComment error. awemeId:{},{}", aweme.getOAwemeId(), ex);
//                    XxlJobLogger.log("SaveAwemeComment error. awemeId:" + aweme.getOAwemeId(), ex);
//                }
//            })).get();
//        }
//
//        //对于评论数>指定数的零关键字更新
//        log.info("开始零关键字搜索..... 评论>指定数的视频数={}", gt200commentsAwemes.size());
//        List<Aweme> awemesgt200 = new ArrayList(gt200commentsAwemes.keySet());
//        ConcurrentHashMap<Aweme, List<AwemeComment>> zeroKeywordAwemeComments = new ConcurrentHashMap<>();
//        s3ForkJoinPool.submit(() -> awemesgt200.parallelStream().forEach(a -> {
//            try {
//                //todo 暂不考虑关键词数量》0且需要更新关键词的情形
//                if (findKeywordByES(a).size() <= 0) {
//                    zeroKeywordAwemeComments.put(a, gt200commentsAwemes.get(a));
//                }
//                return;
//            } catch (Exception ex) {
//                log.error("SaveAwemeComment error.过滤掉关键字不为0的视频 awemeId:{},{}", a.getOAwemeId(), ex);
//
//            }
//        })).get();
//
//        int zeroNums = zeroKeywordAwemeComments.size();
//        log.info("开始处理零关键字视频..... 零关键字视频数={}", zeroNums);
//        List<Aweme> zeroKeywordAwemes = new ArrayList(zeroKeywordAwemeComments.keySet());
//        forkJoinPool.submit(() -> zeroKeywordAwemes.parallelStream().forEach(a -> {
//            try {
//                AwemesComentStatics.AwemeComentStatics awemeComentStatics = new AwemesComentStatics.AwemeComentStatics();
//                awemeComentStatics.setAwemesComentStatics(event.zeroKeyWordStatics);
//                awemeComentStatics.setAwemeId(a.getOAwemeId());
//                // 解析关键字,临时代码
//                try {
//
//                    val saveAwemeKeywordEvent = new SaveAwemeKeywordEventListener.SaveAwemeKeywordEvent(zeroKeywordAwemeComments.get(a), a);
//                    saveAwemeKeywordEvent.setAwemeComentStatics(awemeComentStatics);
//                    eventBusTemplate.post(saveAwemeKeywordEvent);
//                    awemeComentStatics.setCostTimes((System.currentTimeMillis() - start) / 1000);
//                } catch (Exception e) {
//                    e.printStackTrace();
//                } finally {
////                        event.zeroKeyWordStatics.getOnlyAnAwemes().incrementAndGet();
//                    log.info("只解析零关键字：{}/{},{}", event.currZeroKeywordNum.incrementAndGet(), zeroNums, awemeComentStatics);
//                }
//                return;
//            } catch (Exception ex) {
//                log.error("SaveAwemeComment error.解析零关键词的视频 awemeId:{},{}", a.getOAwemeId(), ex);
//
//            }
//        })).get();
//
//        log.info("SaveAwemeCommentEvent real end awemeSize :{}/{}, real zero keyword aweme size={}/{},use :{}s,", event.currNum.get(), event.awemes.size(), event.currZeroKeywordNum.get(), zeroNums, (System.currentTimeMillis() - start) / 1000);
//        XxlJobLogger.log("SaveAwemeCommentEvent real end awemeSize :{}/{}, real zero keyword aweme size={}/{},use :{}s,", event.currNum.get(), event.awemes.size(), event.currZeroKeywordNum.get(), zeroNums, (System.currentTimeMillis() - start) / 1000);
//    }
//
//    private void run(Aweme aweme, SaveAwemeCommentEvent event) {
//
//        long start = System.currentTimeMillis();
//        AwemesComentStatics.AwemeComentStatics awemeComentStatics = new AwemesComentStatics.AwemeComentStatics();
//        awemeComentStatics.setAwemesComentStatics(event.statics);
//        awemeComentStatics.setAwemeId(aweme.getOAwemeId());
//        awemeComentStatics.setOCommentCount(aweme.getOCommentCount());
//        AwemeCommentResponse response;
//        try {
//            response = awemeCommentSpiderClient.get(aweme.getOAwemeId(), 0, 40);
//        } catch (Exception e) {
//            awemeComentStatics.getAwemesComentStatics().getTotalCommentSpideErrors().incrementAndGet();
//            log.error("采集评论异常 {}/{},{},{}", event.currNum.incrementAndGet(), event.statics.getTotalAwemes(), awemeComentStatics, e.getMessage());
//            return;
//        }
//        awemeComentStatics.setCommentSpiderTime((System.currentTimeMillis() - start) / 1000);
//
////        if((System.currentTimeMillis()-start)/1000>2)
////            XxlJobLogger.log("SaveAwemeComment awemeCommentSpiderClient awemeId:{} cursor:{} use time:{}s",aweme.getOAwemeId(),0,(System.currentTimeMillis()-start)/1000);
//        if (CollectionUtils.isEmpty(response.getComments())) {
//            awemeComentStatics.setHasComments(false);
//            awemeComentStatics.setCommentSize(-1);
//            awemeComentStatics.getAwemesComentStatics().getTotalCommentSpideIsEmptys().incrementAndGet();
//            log.info("未采集到评论 {}/{},{}", event.currNum.incrementAndGet(), event.statics.getTotalAwemes(), awemeComentStatics);
//
//            return;
//        }
//        List<AwemeComment> awemeComments = awemeCommentMapper.mapper(response.getComments());
//        //最多200条
//        int needComments = Math.min(aweme.getOCommentCount(), collectMaxComments);
//        while (awemeComments.size() <= needComments && response.isHas_more()) {
//            try {
//                long start2 = System.currentTimeMillis();
//                response = awemeCommentSpiderClient.get(aweme.getOAwemeId(), response.getCursor(), 40);
//
//                if ((System.currentTimeMillis() - start2) / 1000 > 5) {
//                    log.info("SaveAwemeComment awemeCommentSpiderClient awemeId:{} cursor:{} use time:{}s", aweme.getOAwemeId(), response.getCursor(), (System.currentTimeMillis() - start2) / 1000);
//                    XxlJobLogger.log("SaveAwemeComment awemeCommentSpiderClient awemeId:{} cursor:{} use time:{}s", aweme.getOAwemeId(), response.getCursor(), (System.currentTimeMillis() - start2) / 1000);
//                }
//                if (CollectionUtils.isEmpty(response.getComments())) break;
//                awemeComments.addAll(awemeCommentMapper.mapper(response.getComments()));
//            } catch (Exception e) {
//                log.error(e.getMessage(), e);
//                break;
//            }
//        }
//        awemeComentStatics.setCommentSpiderTime((System.currentTimeMillis() - start) / 1000);
//        long s3Start = System.currentTimeMillis();
//        //保存到s3
//        awemeCommentRepository.save(aweme.getOAwemeId(), awemeComments);
//        awemeComentStatics.setCommentSize(awemeComments.size());
//        awemeComentStatics.setSaveComments2S3Time((System.currentTimeMillis() - s3Start) / 1000);
//        if ((System.currentTimeMillis() - start) / 1000 > 5) {
//            log.info("SaveAwemeCommentEvent end awemeId :{} awemeComentSize:{} use :{}s", aweme.getOAwemeId(), awemeComments.size(), (System.currentTimeMillis() - start) / 1000);
//            XxlJobLogger.log("SaveAwemeCommentEvent end awemeId :{} awemeComentSize:{} use :{}s", aweme.getOAwemeId(), awemeComments.size(), (System.currentTimeMillis() - start) / 1000);
//        }
//
//        // redis日志记录
//        /*synchronized (this){
//            LogUtil.addCurSize(1);
//        }*/
//        if (awemeComments.size() > 50) {
//            // 解析关键字
//            val saveAwemeKeywordEvent = new SaveAwemeKeywordEventListener.SaveAwemeKeywordEvent(awemeComments, aweme);
//            saveAwemeKeywordEvent.setAwemeComentStatics(awemeComentStatics);
//            eventBusTemplate.post(saveAwemeKeywordEvent);
//            awemeComentStatics.setCostTimes((System.currentTimeMillis() - start) / 1000);
//            log.info("{}/{},{}", event.currNum.incrementAndGet(), event.statics.getTotalAwemes(), awemeComentStatics);
//        } else {
//            awemeComentStatics.getAwemesComentStatics().getTotalCommentSpideLt50().incrementAndGet();
//            log.info("评论采集不够50 {}/{},{}", event.currNum.incrementAndGet(), event.statics.getTotalAwemes(), awemeComentStatics);
//
//        }
//
//    }
//
//    private List<AwemeComment> getCommentsFromS3(long awemeId) {
//        long start = System.currentTimeMillis();
//        List<AwemeComment> result = new ArrayList<>();
//
//        try {
//            String comment = awemeCommentRepository.get(awemeId);
//
//            if ((System.currentTimeMillis() - start) / 1000 > 5) {
//                log.info("SaveAwemeComment getCommentCount from s3 data awemeId:{} use time:{}s", awemeId, (System.currentTimeMillis() - start) / 1000);
//                XxlJobLogger.log("SaveAwemeComment getCommentCount from  s3 data awemeId:{} use time:{}s", awemeId, (System.currentTimeMillis() - start) / 1000);
//            }
//
//            JavaType jt = MAPPER.getTypeFactory().constructParametricType(ArrayList.class, AwemeComment.class);
//            List<AwemeComment> rst = MAPPER.readValue(comment, jt);
//            return rst;
////            result = JsonUtil.jsonParse(List.class, comment);
//        } catch (Exception e) {
//            // log.error("SaveAwemeComment get s3 data error!",e);
//        }
//        return result;
//    }
//
//    private int getCommentCount(long awemeId) {
//        long start = System.currentTimeMillis();
//        int result = 0;
//
//        try {
//            String comment = awemeCommentRepository.get(awemeId);
//
//            if ((System.currentTimeMillis() - start) / 1000 > 5) {
//                log.info("SaveAwemeComment getCommentCount from s3 data awemeId:{} use time:{}s", awemeId, (System.currentTimeMillis() - start) / 1000);
//                XxlJobLogger.log("SaveAwemeComment getCommentCount from  s3 data awemeId:{} use time:{}s", awemeId, (System.currentTimeMillis() - start) / 1000);
//            }
//            result = JsonUtil.jsonParse(List.class, comment).size();
//        } catch (Exception e) {
//            // log.error("SaveAwemeComment get s3 data error!",e);
//        }
//        return result;
//    }
//
//
//    private List<Keyword> findKeywordByES(Aweme aweme) {
//        GetRequest getRequest = new GetRequest(AWEME_INDEX + DateUtil.format(DateUtil.calendar(aweme.getOCreateTime() * 1000).getTime(), "yyyyMM"),
//                String.valueOf(aweme.getId()));
//        FetchSourceContext fetchSourceContext = new FetchSourceContext(true);
//        getRequest.fetchSourceContext(fetchSourceContext);
//        GetResponse getResponse;
//
//        try {
//            long start = System.currentTimeMillis();
//            getResponse = highLevelClient.get(getRequest, RequestOptions.DEFAULT);
//
//            if ((System.currentTimeMillis() - start) / 1000 > 5) {
//                log.info("SaveAwemeComment findKeywordByEs awemeId:{} use time:{}ms", aweme.getOAwemeId(), (System.currentTimeMillis() - start));
//                XxlJobLogger.log("SaveAwemeComment findKeywordByEs awemeId:{} use time:{}ms", aweme.getOAwemeId(), (System.currentTimeMillis() - start));
//            }
//
//
//        } catch (IOException e) {
//            log.error("ESRefreshAwemeEventListener findEs error!", e);
//            return new ArrayList<>();
//        }
//
//        if (null == getResponse.getSource()) {
//            return new ArrayList<>();
//        }
//
//        if (null == getResponse.getSource().get("awemeTags")) {
//            return new ArrayList<>();
//        }
//
//        return (List<Keyword>) getResponse.getSource().get("awemeTags");
//    }
//
//    @Setter
//    @Getter
//    public static class SaveAwemeCommentEvent extends Event {
//
//        public SaveAwemeCommentEvent(Event event, List<Aweme> awemes) {
//            super(event, "SaveAwemeComment");
//            this.awemes = awemes;
//        }
//
//        public SaveAwemeCommentEvent(List<Aweme> awemes) {
//            super("SaveAwemeComment");
//            this.awemes = awemes;
//        }
//        private String type;
//        private List<Aweme> awemes;
//        private AwemesComentStatics statics = new AwemesComentStatics();
//        private AwemesComentStatics zeroKeyWordStatics = new AwemesComentStatics();
//        private AtomicInteger currNum = new AtomicInteger();
//        private AtomicInteger currZeroKeywordNum = new AtomicInteger();
//        private AtomicInteger commentAnalyzedTotalErrors = new AtomicInteger();
//
//        public void setType(String type) {
//            this.type = type;
//            statics.setType(type);
//            zeroKeyWordStatics.setType(type);
//        }
//    }
//}
